package com.shujia.sql

import org.apache.spark.sql.expressions.Window
import org.apache.spark.sql.{DataFrame, SparkSession}

/**
 * 开窗函数
 * 聚合开窗函数：sum  count avg min max
 * 排序开窗函数：row_number rank desen_rank lag（向上取） lead（向后取）
 */
object Demo6Window {
  def main(args: Array[String]): Unit = {
    //创建SparkSession对象
    /**
     * 在新版本的spark中，如果想要编写spark sql的话，需要使用新的spark入口类：SparkSession
     */
    val sparkSession: SparkSession = SparkSession.builder()
      .master("local")
      .appName("开窗函数DSL API演示")
      .config("spark.sql.shuffle.partitions", 1) //默认分区的数量是200个
      .getOrCreate()

    //导入Spark sql中所有的sql隐式转换函数
    import org.apache.spark.sql.functions._
    //导入另一个隐式转换，后面可以直接使用$函数引用字段进行处理,如果需要做RDD和DF之间的转换
    import sparkSession.implicits._

    //学生表
    val studentsDF: DataFrame = sparkSession.read
      .format("csv")
      .option("sep", ",")
      .schema("id STRING,name STRING,age INT,gender STRING,clazz STRING")
      .load("spark/data/students.csv")

    //成绩表
    val scoresDF: DataFrame = sparkSession.read
      .format("csv")
      .option("sep", ",")
      .schema("sid STRING,subject_id STRING,score INT")
      .load("spark/data/score.txt")

    //科目表
    val subjectDF: DataFrame = sparkSession.read
      .format("csv")
      .option("sep", ",")
      .schema("subject_id STRING,subject_name STRING,subject_sum_score INT")
      .load("spark/data/subject.csv")

    //将学生数据与成绩数据进行关联
    val joinDF: DataFrame = studentsDF.join(scoresDF, $"id" === $"sid")
    //    joinDF.show(10)

    /**
     * 1、统计总分年级排名前十学生各科的分数
     *
     * 未排序之前，是将开窗中所有数据一起聚合得到一个结果
     * 若排序了，依次从上到下聚合得到一个结果
     *
     */
    joinDF
      // sum(score) over(partition by id ) as sumScore
      .withColumn("sumScore", sum($"score") over Window.partitionBy($"id"))
      .orderBy($"sumScore".desc)
      .limit(60)
    //.show(60)

    /**
     * 3、统计每科都及格的学生
     */
    scoresDF
      .join(subjectDF, "subject_id")
      .where($"score" >= $"subject_sum_score" * 0.6)
      //统计学生及格的科目数
      .withColumn("jiGeCounts", count($"sid") over Window.partitionBy($"sid"))
      .where($"jiGeCounts" === 6)
    //      .show(100)

    /**
     * 2、统计总分大于年级平均分的学生
     */
    joinDF
      //计算每个学生的总分，新增一列
      .withColumn("sumScore", sum($"score") over Window.partitionBy($"id"))
      .withColumn("avgScore", avg($"sumScore") over Window.partitionBy(substring($"clazz", 0, 2)))
      .where($"sumScore" > $"avgScore")
    //      .show(200)

    /**
     * 统计每个班级的每个名次之间的分数差
     *
     */
    joinDF
      .groupBy($"id", $"clazz")
      .agg(sum($"score") as "sumScore")
      //开窗，班级开窗，总分降序排序，排个名次
      .withColumn("rn", row_number() over Window.partitionBy($"clazz").orderBy($"sumScore".desc))
      //开窗，取出前一名的总分
      .withColumn("front_score", lag($"sumScore",1,750) over Window.partitionBy($"clazz").orderBy($"sumScore".desc))
      .withColumn("cha",$"front_score" - $"sumScore")
      .show(100)


  }
}
